package stream

import (
	"errors"
	"reflect"
	"stream/optional"
	"sync"
)

// Comparable is an object Comparator
type Comparable[T any] func(T, T) int

// Stream is Defined streaming behavior
type Stream[T any] interface {
	Filter(func(T) bool) Stream[T]
	Peek(func(T)) Stream[T]
	Sorted(Comparable[T]) Stream[T]
	Skip(n int) Stream[T]
	Limit(n int) Stream[T]
	Distinct() Stream[T]
	Parallel() Stream[T]
	ForEach(func(T))
	Count() int
	MapToUint(func(T) uint) Stream[uint]
	MapToUint8(func(T) uint8) Stream[uint8]
	MapToUint16(func(T) uint16) Stream[uint16]
	MapToUint32(func(T) uint32) Stream[uint32]
	MapToUint64(func(T) uint64) Stream[uint64]
	MapToInt(func(T) int) Stream[int]
	MapToInt8(func(T) int8) Stream[int8]
	MapToInt16(func(T) int16) Stream[int16]
	MapToInt32(func(T) int32) Stream[int32]
	MapToInt64(func(T) int64) Stream[int64]
	MapToFloat32(func(T) float32) Stream[float32]
	MapToFloat64(func(T) float64) Stream[float64]
	MapToString(func(T) string) Stream[string]
	MapToBool(func(T) bool) Stream[bool]
	MapToAny(func(T) any) Stream[any]
	MapToComplex64(func(T) complex64) Stream[complex64]
	MapToComplex128(func(T) complex128) Stream[complex128]
	FlatMapToUint(func(T) Stream[uint]) Stream[uint]
	FlatMapToUint8(func(T) Stream[uint8]) Stream[uint8]
	FlatMapToUint16(func(T) Stream[uint16]) Stream[uint16]
	FlatMapToUint32(func(T) Stream[uint32]) Stream[uint32]
	FlatMapToUint64(func(T) Stream[uint64]) Stream[uint64]
	FlatMapToInt(func(T) Stream[int]) Stream[int]
	FlatMapToInt8(func(T) Stream[int8]) Stream[int8]
	FlatMapToInt16(func(T) Stream[int16]) Stream[int16]
	FlatMapToInt32(func(T) Stream[int32]) Stream[int32]
	FlatMapToInt64(func(T) Stream[int64]) Stream[int64]
	FlatMapToFloat32(func(T) Stream[float32]) Stream[float32]
	FlatMapToFloat64(func(T) Stream[float64]) Stream[float64]
	FlatMapToString(func(T) Stream[string]) Stream[string]
	FlatMapToBool(func(T) Stream[bool]) Stream[bool]
	FlatMapToComplex64(func(T) Stream[complex64]) Stream[complex64]
	FlatMapToComplex128(func(T) Stream[complex128]) Stream[complex128]
	FlatMapToAny(func(T) Stream[any]) Stream[any]
	Max(Comparable[T]) *optional.Optional[T]
	Min(Comparable[T]) *optional.Optional[T]
	Reduce(func(T, T) T) *optional.Optional[T]
	Paralleled() bool
	Lazy() bool
	// Closed is checked to stream whether is closed
	Closed() bool
	ToSlice() []T
}

// stream is a Stream implement.
type stream[T any] struct {
	buf      []T
	fns      []func()
	lazy     bool
	parallel bool
	closed   bool
}

// NewStream can create a delay calculation stream.
func NewStream[T any](v []T) Stream[T] {
	return &stream[T]{buf: v, lazy: true}
}

// NewPositiveStream can create a positive calculation stream.
// This means that methods such as Filter, Peek, Sorted, Skip, Limit and Distinct will calculate the results immediately
func NewPositiveStream[T any](v []T) Stream[T] {
	return &stream[T]{buf: v}
}

func (s *stream[T]) Filter(f func(T) bool) Stream[T] {
	fn := func() {
		if len(s.buf) == 0 {
			return
		}
		if s.parallel {
			var dfs func(t []T, left, right int, ch chan []T)
			dfs = func(t []T, left, right int, ch chan []T) {
				if right-left == 1 {
					if f(t[left]) {
						ch <- t[left:right]
					} else {
						ch <- nil
					}
					return
				}
				mid := (left + right) / 2
				ch1, ch2 := make(chan []T), make(chan []T)
				defer func() {
					close(ch1)
					close(ch2)
				}()
				go dfs(t, left, mid, ch1)
				go dfs(t, mid, right, ch2)
				ans1 := <-ch1
				ans2 := <-ch2
				var ans []T
				ans = append(ans, ans1...)
				ans = append(ans, ans2...)
				ch <- ans
			}
			ch := make(chan []T)
			defer close(ch)
			go dfs(s.buf, 0, len(s.buf), ch)
			buf := <-ch
			s.buf = buf
		} else {
			buf := make([]T, 0, len(s.buf))
			for i := 0; i < len(s.buf); i++ {
				if f(s.buf[i]) {
					buf = append(buf, s.buf[i])
				}
			}
			s.buf = buf
		}
	}

	if s.lazy {
		s.fns = append(s.fns, fn)
		return s
	}

	fn()
	return s
}

func (s *stream[T]) Peek(f func(T)) Stream[T] {
	fn := func() {
		if len(s.buf) == 0 {
			return
		}
		if s.parallel {
			var wg sync.WaitGroup
			wg.Add(len(s.buf))
			buf := s.buf
			for _, v := range buf {
				v := v
				go func() {
					f(v)
					wg.Done()
				}()
			}
			wg.Wait()
		} else {
			for _, v := range s.buf {
				f(v)
			}
		}
	}
	if s.lazy {
		s.fns = append(s.fns, fn)
		return s
	}

	fn()
	return s
}

func (s *stream[T]) Sorted(f Comparable[T]) Stream[T] {
	fn := func() {
		if len(s.buf) == 0 {
			return
		}
		if len(s.buf) >= 12 && s.parallel {
			var sf func(s []T, l, r int, ch chan []T)
			sf = func(s []T, l, r int, ch chan []T) {
				if r-l == 1 {
					ch <- s[l:r]
					return
				}
				mid := (l + r) / 2
				ch1, ch2 := make(chan []T), make(chan []T)
				defer func() {
					close(ch1)
					close(ch2)
				}()
				go sf(s, l, mid, ch1)
				go sf(s, mid, r, ch2)
				ans1 := <-ch1
				ans2 := <-ch2
				ans := make([]T, 0, r-l)
				i, j := 0, 0
				for i < len(ans1) && j < len(ans2) {
					if f(ans1[i], ans2[j]) < 0 {
						ans = append(ans, ans1[i])
						i++
					} else {
						ans = append(ans, ans2[j])
						j++
					}
				}
				for ; i < len(ans1); i++ {
					ans = append(ans, ans1[i])
				}
				for ; j < len(ans2); j++ {
					ans = append(ans, ans2[j])
				}
				ch <- ans
			}
			ch := make(chan []T)
			defer close(ch)
			go sf(s.buf, 0, len(s.buf), ch)
			s.buf = <-ch
		} else {
			buf := make([]T, len(s.buf))
			copy(buf, s.buf)
			s.buf = buf
			quickSort(s.buf, 0, len(s.buf), maxDepth(len(s.buf)), f)
		}
	}

	if s.lazy {
		s.fns = append(s.fns, fn)
		return s
	}

	fn()
	return s
}

func (s *stream[T]) Skip(n int) Stream[T] {
	fn := func() {
		s.buf = s.buf[n:]
	}

	if s.lazy {
		s.fns = append(s.fns, fn)
		return s
	}

	fn()
	return s
}

func (s *stream[T]) Limit(n int) Stream[T] {
	fn := func() {
		s.buf = s.buf[:n]
	}

	if s.lazy {
		s.fns = append(s.fns, fn)
		return s
	}

	fn()
	return s
}

func (s *stream[T]) Distinct() Stream[T] {
	fn := func() {
		rmPoint := make(map[int]bool)
		for i := 0; i < len(s.buf); i++ {
			_, exist := rmPoint[i]
			if exist {
				continue
			}
			for j := i + 1; j < len(s.buf); j++ {
				_, exist := rmPoint[j]
				if exist {
					continue
				}
				if reflect.DeepEqual(s.buf[i], s.buf[j]) {
					rmPoint[j] = true
				}
			}
		}
		if len(rmPoint) > 0 {
			buf := make([]T, 0, len(s.buf)-len(rmPoint))
			for i := 0; i < len(s.buf); i++ {
				_, exist := rmPoint[i]
				if !exist {
					buf = append(buf, s.buf[i])
				}
			}
			s.buf = buf
		}
	}

	if s.lazy {
		s.fns = append(s.fns, fn)
		return s
	}

	fn()
	return s
}

func (s *stream[T]) Parallel() Stream[T] {
	s.parallel = true
	return s
}

func (s *stream[T]) ForEach(f func(T)) {
	defer s.close()
	s.fns = append(s.fns, func() {
		buf := s.buf
		for _, v := range buf {
			f(v)
		}
	})
	s.completable()
}

func (s *stream[T]) Count() int {
	defer s.close()
	s.completable()
	return len(s.buf)
}

func (s *stream[T]) Min(f Comparable[T]) *optional.Optional[T] {
	return optional.Of(s.min(f))
}

func (s *stream[T]) min(f Comparable[T]) (T, error) {
	defer s.close()
	s.completable()
	if len(s.buf) == 0 {
		var zero T
		return zero, errors.New("cannot get minimum value with size 0")
	}
	index := 0
	min := s.buf[index]
	for i := 1; i < len(s.buf); i++ {
		if f(min, s.buf[i]) > 0 {
			index = i
			min = s.buf[i]
		}
	}
	return min, nil
}

func (s *stream[T]) Max(f Comparable[T]) *optional.Optional[T] {
	return optional.Of(s.max(f))
}

func (s *stream[T]) max(f Comparable[T]) (T, error) {
	defer s.close()
	s.completable()
	if len(s.buf) == 0 {
		var zero T
		return zero, errors.New("cannot get maximum value with size 0")
	}
	index := 0
	max := s.buf[index]
	for i := 1; i < len(s.buf); i++ {
		if f(max, s.buf[i]) < 0 {
			index = i
			max = s.buf[i]
		}
	}
	return max, nil
}

func (s *stream[T]) Reduce(f func(T, T) T) *optional.Optional[T] {
	return optional.Of(s.reduce(f))
}

func (s *stream[T]) reduce(f func(T, T) T) (T, error) {
	defer s.close()
	s.completable()
	if len(s.buf) == 0 {
		var zero T
		return zero, errors.New("cannot reduce items with size 0")
	}
	total := s.buf[0]
	for i := 1; i < len(s.buf); i++ {
		total = f(total, s.buf[i])
	}
	return total, nil
}

func (s *stream[T]) ToSlice() []T {
	defer s.close()
	s.completable()
	return s.buf
}

func (s *stream[T]) close() {
	s.buf = nil
	s.closed = true
}

func (s *stream[T]) completable() {
	for _, fn := range s.fns {
		fn()
	}
	s.fns = nil
}

func (s *stream[T]) MapToUint(f func(T) uint) Stream[uint] {
	return newMapStream(s, f)
}

func (s *stream[T]) MapToUint8(f func(T) uint8) Stream[uint8] {
	return newMapStream(s, f)
}

func (s *stream[T]) MapToUint16(f func(T) uint16) Stream[uint16] {
	return newMapStream(s, f)
}

func (s *stream[T]) MapToUint32(f func(T) uint32) Stream[uint32] {
	return newMapStream(s, f)
}

func (s *stream[T]) MapToUint64(f func(T) uint64) Stream[uint64] {
	return newMapStream(s, f)
}

func (s *stream[T]) MapToInt(f func(T) int) Stream[int] {
	return newMapStream(s, f)
}

func (s *stream[T]) MapToInt8(f func(T) int8) Stream[int8] {
	return newMapStream(s, f)
}

func (s *stream[T]) MapToInt16(f func(T) int16) Stream[int16] {
	return newMapStream(s, f)
}

func (s *stream[T]) MapToInt32(f func(T) int32) Stream[int32] {
	return newMapStream(s, f)
}

func (s *stream[T]) MapToInt64(f func(T) int64) Stream[int64] {
	return newMapStream(s, f)
}

func (s *stream[T]) MapToFloat32(f func(T) float32) Stream[float32] {
	return newMapStream(s, f)
}

func (s *stream[T]) MapToFloat64(f func(T) float64) Stream[float64] {
	return newMapStream(s, f)
}

func (s *stream[T]) MapToString(f func(T) string) Stream[string] {
	return newMapStream(s, f)
}

func (s *stream[T]) MapToBool(f func(T) bool) Stream[bool] {
	return newMapStream(s, f)
}

func (s *stream[T]) MapToComplex64(f func(T) complex64) Stream[complex64] {
	return newMapStream(s, f)
}

func (s *stream[T]) MapToComplex128(f func(T) complex128) Stream[complex128] {
	return newMapStream(s, f)
}

func (s *stream[T]) MapToAny(f func(T) any) Stream[any] {
	return newMapStream(s, f)
}

func (s *stream[T]) FlatMapToUint(f func(T) Stream[uint]) Stream[uint] {
	return newFlatMapStream(s, f)
}

func (s *stream[T]) FlatMapToUint8(f func(T) Stream[uint8]) Stream[uint8] {
	return newFlatMapStream(s, f)
}

func (s *stream[T]) FlatMapToUint16(f func(T) Stream[uint16]) Stream[uint16] {
	return newFlatMapStream(s, f)
}

func (s *stream[T]) FlatMapToUint32(f func(T) Stream[uint32]) Stream[uint32] {
	return newFlatMapStream(s, f)
}

func (s *stream[T]) FlatMapToUint64(f func(T) Stream[uint64]) Stream[uint64] {
	return newFlatMapStream(s, f)
}

func (s *stream[T]) FlatMapToInt(f func(T) Stream[int]) Stream[int] {
	return newFlatMapStream(s, f)
}

func (s *stream[T]) FlatMapToInt8(f func(T) Stream[int8]) Stream[int8] {
	return newFlatMapStream(s, f)
}

func (s *stream[T]) FlatMapToInt16(f func(T) Stream[int16]) Stream[int16] {
	return newFlatMapStream(s, f)
}

func (s *stream[T]) FlatMapToInt32(f func(T) Stream[int32]) Stream[int32] {
	return newFlatMapStream(s, f)
}

func (s *stream[T]) FlatMapToInt64(f func(T) Stream[int64]) Stream[int64] {
	return newFlatMapStream(s, f)
}

func (s *stream[T]) FlatMapToFloat32(f func(T) Stream[float32]) Stream[float32] {
	return newFlatMapStream(s, f)
}

func (s *stream[T]) FlatMapToFloat64(f func(T) Stream[float64]) Stream[float64] {
	return newFlatMapStream(s, f)
}

func (s *stream[T]) FlatMapToString(f func(T) Stream[string]) Stream[string] {
	return newFlatMapStream(s, f)
}

func (s *stream[T]) FlatMapToBool(f func(T) Stream[bool]) Stream[bool] {
	return newFlatMapStream(s, f)
}

func (s *stream[T]) FlatMapToComplex64(f func(T) Stream[complex64]) Stream[complex64] {
	return newFlatMapStream(s, f)
}

func (s *stream[T]) FlatMapToComplex128(f func(T) Stream[complex128]) Stream[complex128] {
	return newFlatMapStream(s, f)
}

func (s *stream[T]) FlatMapToAny(f func(T) Stream[any]) Stream[any] {
	return newFlatMapStream(s, f)
}

func (s *stream[T]) Closed() bool {
	return s.closed
}

func (s *stream[T]) Paralleled() bool {
	return s.parallel
}

func (s *stream[T]) Lazy() bool {
	return s.lazy
}

func Map[T any, V any](s Stream[T], f func(T) V) *optional.Optional[Stream[V]] {
	ns, ok := s.(*stream[T])
	if !ok {
		return optional.Of[Stream[V]](nil, errors.New("this method only supports parameters of type *stream[T] for instances"))
	}
	return optional.Of(newMapStream(ns, f), nil)
}

func FlatMap[T any, V any](s Stream[T], f func(T) Stream[V]) *optional.Optional[Stream[V]] {
	ns, ok := s.(*stream[T])
	if !ok {
		return optional.Of[Stream[V]](nil, errors.New("this method only supports parameters of type *stream[T] for instances"))
	}
	return optional.Of(newFlatMapStream(ns, f), nil)
}

func newMapStream[T any, V any](s *stream[T], f func(T) V) Stream[V] {
	var ret Stream[V]
	var fn func()
	if s.parallel {
		fn = func() {
			var buf []V
			if len(s.buf) == 0 {
				buf = make([]V, 0)
			} else {
				var dfs func(s []T, l, r int, ch chan []V)
				dfs = func(s []T, l, r int, ch chan []V) {
					if r-l == 1 {
						ch <- []V{f(s[l])}
						return
					}
					ch1, ch2 := make(chan []V), make(chan []V)
					defer func() {
						close(ch1)
						close(ch2)
					}()
					mid := (l + r) / 2
					go dfs(s, l, mid, ch1)
					go dfs(s, mid, r, ch2)
					ans1 := <-ch1
					ans2 := <-ch2
					ans := make([]V, 0, r-l)
					ans = append(ans, ans1...)
					ans = append(ans, ans2...)
					ch <- ans
				}
				ch := make(chan []V)
				defer close(ch)
				go dfs(s.buf, 0, len(s.buf), ch)
				buf = <-ch
			}
			ret.(*stream[V]).buf = buf
		}
	} else {
		fn = func() {
			buf := make([]V, 0, len(s.buf))
			for _, v := range s.buf {
				buf = append(buf, f(v))
			}
			ret.(*stream[V]).buf = buf
		}
	}
	if s.lazy {
		ret = NewStream[V](nil)
		ret.(*stream[V]).fns = append(s.fns, fn)
		if s.parallel {
			ret = ret.Parallel()
		}
	} else {
		ret = NewPositiveStream[V](nil)
		if s.parallel {
			ret = ret.Parallel()
		}
		fn()
	}
	return ret
}

func newFlatMapStream[T any, V any](s *stream[T], f func(T) Stream[V]) Stream[V] {
	// todo parallel implements
	var ret Stream[V]
	var fn func()
	fn = func() {
		buf := make([]V, 0, len(s.buf)*2)
		for _, v := range s.buf {
			buf = append(buf, f(v).(*stream[V]).buf...)
		}
		ret.(*stream[V]).buf = buf
	}

	if s.lazy {
		ret = NewStream[V](nil)
		ret.(*stream[V]).fns = append(s.fns, fn)
		if s.parallel {
			ret = ret.Parallel()
		}
	} else {
		ret = NewPositiveStream[V](nil)
		if s.parallel {
			ret = ret.Parallel()
		}
		fn()
	}

	return ret
}
